package com.group.sliding.utils;

import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.group.sliding.context.AdminContext;
import com.group.sliding.domain.po.Item;
import com.group.sliding.domain.po.ItemDoc;
import com.group.sliding.domain.query.ItemPageQuery;
import com.group.sliding.domain.vo.ItemVO;
import com.group.sliding.domain.vo.SuggestVO;
import com.group.sliding.enums.ItemADFlag;
import com.group.sliding.enums.ItemStatus;
import com.group.sliding.exception.BizIllegalException;
import com.group.sliding.properties.ESProperties;
import com.group.sliding.result.PageDTO;
import com.group.sliding.service.IItemService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.client.config.RequestConfig;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.CountRequest;
import org.elasticsearch.client.core.CountResponse;
import org.elasticsearch.common.lucene.search.function.CombineFunction;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.functionscore.FunctionScoreQueryBuilder;
import org.elasticsearch.index.query.functionscore.ScoreFunctionBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
@Component
@RequiredArgsConstructor
public class ESUtils {
    private final IItemService itemService;

    private final RestHighLevelClient client;
    private final ESProperties properties;
    private static final String ERROR_MESSAGE = "elasticsearch error";
    private static final RequestOptions REQUEST_OPTIONS;
    private static final int BULK_SIZE = 1000;
    private static final String KEY_FIELD = "name";
    private static final String HIGHLIGHT_FIELD = "name";
    private static final String STATUS_FIELD = "status";
    private static final String AD_FIELD = "isAD";
    private static final String SUGGESTION_FIELD = "suggestInfo";
    private static final String SUGGESTION_NAME = "suggest";

    static {
        RequestConfig requestConfig = RequestConfig.custom()
                .setConnectTimeout(5000)
                .setSocketTimeout(60000) // 超时时间默认30秒，设为1分钟
                .build();
        REQUEST_OPTIONS = RequestOptions.DEFAULT.toBuilder()
                .setRequestConfig(requestConfig)
                .build();
    }

    /**
     * 批量保存文档
     *
     * @param itemDocs 文档对象列表
     */
    public void saveBatch(List<ItemDoc> itemDocs) {
        try (
                BulkProcessor bulkProcessor = BulkProcessor.builder(
                                (request, bulkListener) -> client.bulkAsync(request, REQUEST_OPTIONS, bulkListener),
                                new BulkProcessor.Listener() {
                                    @Override
                                    public void beforeBulk(long l, BulkRequest request) {
                                        log.info("操作 {} 条数据", request.numberOfActions());
                                    }

                                    @Override
                                    public void afterBulk(long l, BulkRequest request, BulkResponse response) {
                                        log.info("成功 {} 条数据，用时 {}", request.numberOfActions(), response.getTook());
                                    }

                                    @Override
                                    public void afterBulk(long l, BulkRequest request, Throwable throwable) {
                                        log.error("失败 {} 条数据", request.numberOfActions(), throwable);
                                    }
                                })
                        .setBulkActions(BULK_SIZE) // 达到BULK_SIZE的数据量发一次
                        .setBulkSize(new ByteSizeValue(5L, ByteSizeUnit.MB))
                        .setFlushInterval(TimeValue.timeValueSeconds(10L))
                        .setConcurrentRequests(2) // 2线程
                        .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueMillis(100), 3))
                        .build()
        ) {
            if (itemDocs == null || itemDocs.isEmpty()) return;
            log.info("执行新增文档批处理操作...");
            itemDocs.forEach(itemDoc -> {
                IndexRequest indexRequest = new IndexRequest(properties.indexName)
                        .id(itemDoc.getId().toString())
                        .source(JSON.toJSONString(itemDoc), XContentType.JSON);
                // 添加处理，到达BULK_SIZE后自动发送批处理
                bulkProcessor.add(indexRequest);
            });
            // 最后剩余处理手动发送
            bulkProcessor.flush();
            log.info("批量新增文档成功");
        } catch (Exception e) {
            log.error(ERROR_MESSAGE, e);
            throw new BizIllegalException(ERROR_MESSAGE);
        }
    }

    /**
     * 保存文档
     *
     * @param itemDoc 文档对象
     */
    public void save(ItemDoc itemDoc) {
        try {
            log.info("新增文档... {}", itemDoc);
            IndexRequest request = new IndexRequest(properties.indexName)
                    .id(itemDoc.getId().toString())
                    .source(JSON.toJSONString(itemDoc), XContentType.JSON);
            client.index(request, REQUEST_OPTIONS);
            log.info("新增文档成功 {}", itemDoc);
        } catch (IOException e) {
            log.error(ERROR_MESSAGE, e);
            throw new BizIllegalException(ERROR_MESSAGE);
        }
    }

    /**
     * 根据文档id获取文档对象
     *
     * @param id 文档id
     * @return 文档id对应的文档对象
     */
    public ItemDoc getById(String id) {
        try {
            log.info("根据id获取文档信息... id: {}", id);
            GetResponse documentFields = client.get(new GetRequest(properties.getIndexName(), id), REQUEST_OPTIONS);
            log.info("获取到文档信息 {}", documentFields.getSourceAsString());
            return JSON.parseObject(documentFields.getSourceAsString(), ItemDoc.class);
        } catch (IOException e) {
            log.error(ERROR_MESSAGE, e);
            throw new BizIllegalException(ERROR_MESSAGE);
        }
    }

    /**
     * 根据查询条件获取查询结果
     *
     * @param query 查询条件
     * @return 查询结果
     */
    public PageDTO<ItemVO> search(ItemPageQuery query) {
        log.info("执行条件查询... query: {}", query);
        PageDTO<ItemVO> searchResult = new PageDTO<>();
        if (query.getId() != null) {
            // id查询，就查一个
            ItemDoc itemDoc = getById(query.getId().toString());
            searchResult.setTotal(1L);
            searchResult.setPages(1L);
            searchResult.setData(Collections.singletonList(itemDoc.toItemVO(null)));
            return searchResult;
        }
        SearchRequest searchRequest = new SearchRequest(properties.getIndexName());
        SearchSourceBuilder source = searchRequest.source();
        // 布尔查询
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
        // 非管理员不能查到已删除的商品
        if (!AdminContext.isAdmin()) {
            boolQueryBuilder.mustNot().add(QueryBuilders.termQuery(STATUS_FIELD, ItemStatus.DELETED.getValue()));
        }
        if (StrUtil.isNotBlank(query.getKey())) {
            // 关键字查询
            // 推广商品需要往前排，使用算分查询
            boolQueryBuilder.should().add(
                    QueryBuilders.functionScoreQuery(
                            QueryBuilders.matchQuery(KEY_FIELD, query.getKey()), // 查关键字
                            new FunctionScoreQueryBuilder.FilterFunctionBuilder[]{
                                    new FunctionScoreQueryBuilder.FilterFunctionBuilder(
                                            QueryBuilders.termQuery(AD_FIELD, ItemADFlag.IS_AD.getValue()), // 是广告的
                                            ScoreFunctionBuilders.weightFactorFunction(10) // 有10的权重
                                    )
                            }
                    ).boostMode(CombineFunction.MULTIPLY) // 权重和初始的得分值乘算得到最后得分
            );
            // 关键字也可能是查品牌或种类
            boolQueryBuilder.should().add(
                    QueryBuilders.termQuery("brand", query.getKey())
            );
            boolQueryBuilder.should().add(
                    QueryBuilders.termQuery("category", query.getKey())
            );
            // 高亮字段
            source.highlighter(new HighlightBuilder().field(HIGHLIGHT_FIELD).requireFieldMatch(false));
        } else {
            // 全查
            boolQueryBuilder.must().add(QueryBuilders.matchAllQuery());
        }
        // 分页查询
        int from = (query.getPageNo() - 1) * query.getPageSize();
        source.from(from);
        if (from + query.getPageSize() >= 10000) {
            // 超过了elasticsearch的最大查询位置，剩多少查多少
            source.size(10000 - from);
        } else {
            // 正常一页数量
            source.size(query.getPageSize());
        }
        // 价格范围
        if (query.getMaxPrice() != null) {
            boolQueryBuilder.must().add(QueryBuilders.rangeQuery("price").lte(query.getMaxPrice()));
        }
        if (query.getMinPrice() != null) {
            boolQueryBuilder.must().add(QueryBuilders.rangeQuery("price").gte(query.getMinPrice()));
        }
        // 品牌
        if (!StrUtil.isBlank(query.getBrand())) {
            boolQueryBuilder.must().add(QueryBuilders.termQuery("brand", query.getBrand()));
        }
        // 分类
        if (!StrUtil.isBlank(query.getCategory())) {
            boolQueryBuilder.must().add(QueryBuilders.termQuery("category", query.getCategory()));
        }
        // 添加到查询
        source.query(boolQueryBuilder);
        log.info("构建条件查询source: {}", source);
        SearchResponse searchResponse;
        try {
            searchResponse = client.search(searchRequest, REQUEST_OPTIONS);
        } catch (IOException e) {
            log.error(ERROR_MESSAGE, e);
            throw new BizIllegalException(ERROR_MESSAGE);
        }
        // 解析结果
        // 看是否有数据
        SearchHits hits = searchResponse.getHits();
        if (hits == null || hits.getHits().length == 0 || hits.getTotalHits() == null) {
            return searchResult;
        }
        // 总数
        searchResult.setTotal(hits.getTotalHits().value);
        searchResult.setPages(hits.getTotalHits().value / query.getPageSize());
        // 数据
        List<ItemVO> items = new ArrayList<>();
        for (SearchHit hit : hits.getHits()) {
            ItemDoc itemDoc = JSON.parseObject(hit.getSourceAsString(), ItemDoc.class);
            ItemVO itemVO;
            if (hit.getHighlightFields() != null && hit.getHighlightFields().get(HIGHLIGHT_FIELD) != null) {
                // 有高亮的话需要初始化高亮属性
                itemVO = itemDoc.toItemVO(hit.getHighlightFields().get(HIGHLIGHT_FIELD).getFragments()[0].toString());
            } else {
                itemVO = itemDoc.toItemVO(null);
            }
            items.add(itemVO);
        }
        searchResult.setData(items);
        log.info("获取到条件查询结果 {}", searchResult);
        return searchResult;
    }

    /**
     * 根据输入的字符串获取自动补全信息
     *
     * @param prefix 输入
     * @param size   补全信息总数上限
     * @return 补全信息列表
     */
    public List<SuggestVO> suggest(String prefix, int size) {
        log.info("获取补全信息... prefix: {}, size: {}", prefix, size);
        SearchRequest searchRequest = new SearchRequest(properties.getIndexName());
        SearchSourceBuilder source = searchRequest.source();
        source.suggest(
                new SuggestBuilder()
                        .addSuggestion(
                                SUGGESTION_NAME,
                                SuggestBuilders.completionSuggestion(SUGGESTION_FIELD)
                                        .prefix(prefix)
                                        .skipDuplicates(true)
                                        .size(size)
                        )
        );
        log.info("构建补全信息source: {}", source);
        try {
            List<SuggestVO> result = client.search(searchRequest, REQUEST_OPTIONS)
                    .getSuggest()
                    .getSuggestion(SUGGESTION_NAME)
                    .getEntries()
                    .get(0)
                    .getOptions()
                    .stream()
                    .map(Suggest.Suggestion.Entry.Option::getText)
                    .map(Text::toString)
                    .map(SuggestVO::new)
                    .collect(Collectors.toList());
            log.info("获取到补全信息 {}", result);
            return result;
        } catch (Exception e) {
            log.error(ERROR_MESSAGE, e);
            throw new BizIllegalException(ERROR_MESSAGE);
        }
    }

    /**
     * 删除es指定索引库中所有文档
     */
    public void deleteAll() {
        log.info("删除elasticsearch {} 索引库中的所有文档...", properties.getIndexName());
        DeleteByQueryRequest deleteByQueryRequest = new DeleteByQueryRequest(properties.getIndexName());
        deleteByQueryRequest.setQuery(QueryBuilders.matchAllQuery());
        try {
            BulkByScrollResponse response = client.deleteByQuery(deleteByQueryRequest, REQUEST_OPTIONS);
            log.info("删除文档数量: {}", response.getDeleted());
        } catch (IOException e) {
            log.error(ERROR_MESSAGE, e);
            throw new BizIllegalException(ERROR_MESSAGE);
        }
        log.info("成功删除elasticsearch {} 索引库中的所有文档", properties.getIndexName());
    }

    /**
     * 强制用mysql的数据同步elasticsearch的数据，即先删除所有es库中原有数据，然后导入mysql数据
     */
    public void syncElasticsearchByMysql() {
        log.info("开始同步mysql数据到elasticsearch...");
        deleteAll();
        List<Item> list = itemService.list();
        List<ItemDoc> itemDocs = list.stream().map(ItemDoc::fromItem).collect(Collectors.toList());
        saveBatch(itemDocs);
        log.info("成功同步mysql数据到elasticsearch");
    }

    /**
     * 获取文档数量
     */
    public long getCounts() {
        log.info("获取文档总数...");
        CountRequest countRequest = new CountRequest(properties.getIndexName());
        countRequest.query(QueryBuilders.matchAllQuery());
        try {
            CountResponse response = client.count(countRequest, REQUEST_OPTIONS);
            log.info("获取到文档总数: {}", response.getCount());
            return response.getCount();
        } catch (IOException e) {
            log.error(ERROR_MESSAGE, e);
            throw new BizIllegalException(ERROR_MESSAGE);
        }
    }
}
